package com.androwit.engine.cassandra;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.beans.Rows;
import me.prettyprint.hector.api.query.QueryResult;

import com.androwit.engine.MappingReader;
import com.androwit.engine.mapping.MappingField;
import com.androwit.engine.mapping.MultipleColumnField;

public class HectorReader<K, N, V> implements MappingReader<N, V> {
	public static final Object KEY = new Object();
	protected QueryResult<? extends Rows<K, N, V>> queryResult;
	protected Row<K, N, V> currentRow;
	private Iterator<Row<K, N, V>> rowsIterator;
	private ColumnSlice<N, V> columnSlice;
	private List<HColumn<N, V>> columns;
	private HashMap<N, HColumn<N, V>> columnsCache;

	public HectorReader(QueryResult<? extends Rows<K, N, V>> queryResult) {
		this.queryResult = queryResult;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public Map getRelevantValues() {
		Map toReturn = new HashMap();
		toReturn.put(KEY, currentRow.getKey());
		return toReturn;
	}

	public void cacheColumns() {
		columnSlice = currentRow.getColumnSlice();
		columns = columnSlice.getColumns();
		HashMap<N, HColumn<N, V>> toReturn = new HashMap<N, HColumn<N, V>>();
		for (HColumn<N, V> hColumn : columns) {
			toReturn.put(hColumn.getName(), hColumn);
		}
		columnsCache = toReturn;
	}

	@Override
	public boolean readRecord() {
		if (rowsIterator == null) {
			rowsIterator = queryResult.get().iterator();
		}

		if (rowsIterator.hasNext()) {
			currentRow = rowsIterator.next();
			columnsCache = null;
			return true;
		}
		return false;
	}

	@SuppressWarnings("rawtypes")
	public V get(HectorMappingField<?, N, V> column) {
		if (columnsCache == null) {
			cacheColumns();
		}

		if (column instanceof MultipleColumnField)
			return readMultiColumnValue((MultipleColumnField) column);

		return unserializeValue(column);
	}

	/**
	 * 
	 * unserializes Value
	 * 
	 * @param column
	 * @return
	 */
	protected V unserializeValue(HectorMappingField<?, N, V> column) {
		HColumn<N, V> hColumn = columnsCache.get(column.getKey());
		if (hColumn == null)
			return handleValueNotFound(column);

		return column.getValueSerializer().fromByteBuffer(
				hColumn.getValueBytes());
	}

	/**
	 * handles the case a value is not found specifically
	 * 
	 * @param column
	 *            the column definition
	 */
	protected V handleValueNotFound(HectorMappingField<?, N, V> column) {
		// FIXME remove the comment
		// throw new IllegalStateException("Cannot find column " + column +
		// " in row " + currentRow.getKey());
		return null;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	protected V readMultiColumnValue(MultipleColumnField column) {
		return (V) column.parseValues(columnsCache);
	}

	@Override
	public V get(MappingField<?, N, V> field) {
		return get((HectorMappingField<?, N, V>) field);
	}

}
